package com.demo.kafka;


import com.demo.serializer.ISerializer;
import com.demo.serializer.SerializerFactory;
import com.demo.serializer.SerializerType;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.context.SmartLifecycle;
import java.util.*;

public class KafkaSubscriberServer extends KafkaSubscriber implements SmartLifecycle {

    @Override
    public void process(List<ConsumerRecord<String, byte[]>> buffer) throws Exception {
        for (ConsumerRecord<String, byte[]> str : buffer) {
            ISerializer serializer = SerializerFactory.getSerializer(SerializerType.FST);
            Map<String,Object> map= serializer.toObject(str.value(), Map.class);
            throw new RuntimeException("map");
        }
    }

    @Override
    public void processError(List<ConsumerRecord<String, byte[]>> buffer, Exception e) {
        for (ConsumerRecord<String, byte[]> str : buffer) {
            ISerializer serializer = SerializerFactory.getSerializer(SerializerType.FST);
            Map<String,Object> map= serializer.toObject(str.value(), Map.class);
        }
    }

    @Override
    public boolean isAutoStartup() {
        //自动执行start方法
        return true;
    }

    @Override
    public void stop(Runnable runnable) {
        //关闭时执行的操作
    }

    @Override
    public int getPhase() {
        //执行顺序越小越靠前
        return 0;
    }
}